package com.itqiqi.api.tableapi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

public class TableTest4_kafkaPipeLine {

    public static void main(String[] args) throws Exception {

        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO: 2022/5/18 连接kafka读取数据
        tableEnv.connect(new Kafka()
                        .version("0.11")
                        .topic("sensor")
                        .property("zookeeper.connect", "192.168.96.96:2181")
                        .property("bootstrap.servers", "192.168.96.96:9092")
                )
                .withFormat(new Csv())
                .withSchema(new Schema()
                        .field("id", DataTypes.STRING())
                        .field("timestamp", DataTypes.BIGINT())
                        .field("temp", DataTypes.DOUBLE())
                )
                .createTemporaryTable("sensorTable");

        Table sensorTable = tableEnv.from("sensorTable");

        // 简单转化
        Table resTable = sensorTable.select("id, temp").filter("id = 'sensor_1'");
        // 聚合统计
        Table aggTable = sensorTable.groupBy("id").select("id, id.count as cnt, temp.avg as avgTemp");

        // TODO: 2022/5/18 连接kafka输出数据
        tableEnv.connect(new Kafka()
                        .version("0.11")
                        .topic("sinktest")
                        .property("zookeeper.connect", "192.168.96.96:2181")
                        .property("bootstrap.servers", "192.168.96.96:9092")
                )
                .withFormat(new Csv())
                .withSchema(new Schema()
                        .field("id", DataTypes.STRING())
                        .field("temp", DataTypes.DOUBLE())
                )
                .createTemporaryTable("sinktestTable");

        resTable.insertInto("sinktestTable");

        env.execute();

    }
}
